package com.wu.kafka;

import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Properties;

/**
 * @author wufei
 * @date 2018/3/21/021 18:55
 */
public class KafkaProduct {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProduct.class);

    private final static String BORKER_LIST = "47.98.227.22:9092,47.98.186.174:9092";

    KafkaProducer<Integer,String> kafkaProducer;

    private String topic = "part5";

    public KafkaProduct() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BORKER_LIST);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG,"productDemo");
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.wu.kafka.MyPartition");
        kafkaProducer = new KafkaProducer<>(properties);
    }

    public void sendMsg(int key,String msg){
        kafkaProducer.send(new ProducerRecord<>(topic, key, msg),((recordMetadata, e) -> {
            logger.info(recordMetadata.topic()+" "+recordMetadata.offset()+" "+recordMetadata.partition());
        }));
    }

}
